package com.bff.gaia.mix.api.xjoin2.functions;

import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.mix.api.functions.async.ResultFuture;
import com.bff.gaia.mix.api.functions.async.RichAsyncFunction;
import com.bff.gaia.mix.api.xjoin2.core.cache.AbsAsyncXjoinCache;
import com.bff.gaia.mix.api.xjoin2.core.cache.AsyncCacheObject;
import com.bff.gaia.mix.api.xjoin2.core.cache.LRUAsyncXjoinCache;
import com.bff.gaia.mix.api.xjoin2.core.enums.AsyncXjoinCacheType;
import com.bff.gaia.mix.api.xjoin2.core.enums.XjoinType;
import com.bff.gaia.mix.api.xjoin2.core.table.JoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.XjoinInfo;
import com.bff.gaia.types.Row;

import java.util.Collections;

public abstract class RichAsyncXjoinFunction extends RichAsyncFunction<Row, Row> implements XjoinFillData {
	protected XjoinInfo xjoinInfo;
	protected AbsAsyncXjoinCache xjoinCache;

	public RichAsyncXjoinFunction(XjoinInfo xjoinInfo) {
		this.xjoinInfo = xjoinInfo;
	}

	private void initCache() {
		JoinInfo joinInfo = xjoinInfo.getJoinInfo();
		if (joinInfo.getAsyncCacheType() == null) {
			return;
		} else if (AsyncXjoinCacheType.NONE == joinInfo.getAsyncCacheType()) {
			return;
		} else if (AsyncXjoinCacheType.LRU == joinInfo.getAsyncCacheType()) {
			xjoinCache = new LRUAsyncXjoinCache(joinInfo);
		} else {
			throw new RuntimeException("not support async xjoin cache with type:" + joinInfo.getAsyncCacheType());
		}
		xjoinCache.initCache();
	}

	protected AsyncCacheObject get(String key) {
		return xjoinCache.get(key);
	}

	protected void put(String key, AsyncCacheObject cacheObject) {
		xjoinCache.put(key, cacheObject);
	}

	protected boolean isCacheOpen() {
		return xjoinCache != null;
	}

	protected void dealMissKey(Row input, ResultFuture<Row> resultFuture) {
		if (xjoinInfo.getJoinInfo().getXjoinType() == XjoinType.LEFT) {
			Row row = fillData(input, null);
			resultFuture.complete(Collections.singleton(row));
		} else {
			resultFuture.complete(null);
		}
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		initCache();
	}

	@Override
	public void close() throws Exception {
		super.close();
	}
}